package org.duang.db.elasticsearch;

import static org.elasticsearch.common.settings.Settings.settingsBuilder;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang3.StringUtils;
import org.duang.db.Connection;
import org.duang.kit.ToolsKit;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.transport.netty.NettyTransport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ElasticSearchConnection extends Connection {

	private static Logger logger = LoggerFactory.getLogger(ElasticSearchConnection.class);

	protected ElasticSearchConnection searchConn; // 连接对象
	private List<InetSocketTransportAddress> replicaSet; // 集群地址集
	private Settings settings;
	protected Client client; // 客户端对象
	protected String indexname;

	/**
	 * 连接ElasticSearch数据库
	 * 
	 * @param optionMap
	 *            连接参数Map
	 * @throws Exception
	 */
	public void connect(Map<String, String> optionMap) throws Exception {
		initConnect(optionMap); // 初始化连接参数
	}

	@SuppressWarnings("resource")
	private void initConnect(Map<String, String> optionsMap) throws Exception {

		if (null == optionsMap || optionsMap.isEmpty()) {
			throw new NullPointerException("connection mongodb options map is null...");
		}

		database = optionsMap.get(DATABASE_FIELD);
		if (ToolsKit.isEmpty(database)) {
			throw new NullPointerException("elasticsearch cluster name is null!");
		}

		indexname = optionsMap.get(INDEX_NAME_FIELD);

		replicaset = optionsMap.get(REPLICASET_FIELD);
		if (ToolsKit.isEmpty(replicaset)) {
			throw new NullPointerException("elasticsearch replicaset is null!");
		}
		createReplics();
			Map<String,String> settingMap = new HashMap<String,String>();
			settingMap.put("cluster.name", database);
			settingMap.put("action.auto_create_index", "false");
			settingMap.put(NettyTransport.PING_SCHEDULE, "10s"); // enable by default the transport schedule ping interval
			settingMap.put("network.server", "false");
			settingMap.put("node.client", "true");
			  if(replicaSet.size() > 1) {
				  settingMap.put("client.transport.sniff", "true"); //嗅探整个集群的状态，把集群中其它机器的ip地址加到客户端中
			  }
			settingMap.put("cluster.routing.schedule", "500ms");
		  settings = settingsBuilder().put(settingMap).build();  
		  client = TransportClient.builder().settings(settings).build().addTransportAddresses(replicaSet.toArray(new InetSocketTransportAddress[replicaSet.size()]));
	}

	private void createReplics() {
		String[] replicsArray = StringUtils.split(replicaset, ",");
		if (null != replicsArray && replicsArray.length > 0) {
			replicaSet = new ArrayList<InetSocketTransportAddress>(replicsArray.length);
			for (String replic : replicsArray) {
				String[] replicItem = StringUtils.split(replic, ":");
				if (null == replicItem || replicItem.length != 2) {
					throw new RuntimeException("replicsSet is null or length != 2 ");
				}
				int portItem = Integer.parseInt(replicItem[1]);
				if (portItem <= 0)
					throw new RuntimeException("port not is number or <= 0!");
				InetSocketTransportAddress ista = null;
				try {
					ista = new InetSocketTransportAddress(InetAddress.getByName(replicItem[0]), portItem);
				} catch (UnknownHostException e) {
					e.printStackTrace();
				}
				logger.info("host: " + ista.address().getHostName() + "           port: " + ista.address().getPort());
				replicaSet.add(ista);
			}
		}
	}

	@Override
	protected void initDasebase() {
	}

	@Override
	protected void initAuth() {
	}

}
